use async_trait::async_trait;
use thiserror::Error;
use tokio::sync::mpsc;
use std::collections::HashMap;
use std::sync::{Arc, RwLock};
use std::time::{Duration, SystemTime};
use tokio::time;
use tracing::{debug, error, info, warn};
use std::any::Any;

use super::membership::{Member, MemberStatus, MembershipEvent};

#[derive(Debug, Clone)]
pub struct ClusterConfig {
    pub name: String,
    pub host: String,
    pub port: u16,
    /// Heartbeat interval in seconds
    pub heartbeat_interval: u64,
    /// Timeout for marking a node as unreachable (in seconds)
    pub unreachable_timeout: u64,
    /// Timeout for marking a node as down (in seconds)
    pub down_timeout: u64,
}

impl Default for ClusterConfig {
    fn default() -> Self {
        Self {
            name: "default".to_string(),
            host: "127.0.0.1".to_string(),
            port: 8000,
            heartbeat_interval: 5,
            unreachable_timeout: 15,
            down_timeout: 30,
        }
    }
}

#[derive(Error, Debug)]
pub enum ClusterError {
    #[error("Provider error: {0}")]
    ProviderError(String),
    
    #[error("Network error: {0}")]
    NetworkError(String),
    
    #[error("Configuration error: {0}")]
    ConfigError(String),

    #[error("Member not found: {0}")]
    MemberNotFound(String),
}

/// Interface for cluster providers that handle membership discovery
/// and member status tracking
#[async_trait]
pub trait ClusterProvider: Send + Sync + Any {
    /// Register the current node with the cluster
    async fn register(&self, config: &ClusterConfig) -> Result<(), ClusterError>;
    
    /// Deregister the current node from the cluster
    async fn deregister(&self) -> Result<(), ClusterError>;
    
    /// Get the list of all cluster members
    async fn get_members(&self) -> Result<Vec<Member>, ClusterError>;
    
    /// Monitor membership events
    async fn monitor_members(&self) -> mpsc::Receiver<MembershipEvent>;
    
    /// Get self as Any for downcasting
    fn as_any(&self) -> &dyn Any;
}

/// A simple in-memory cluster provider for testing
pub struct MemoryProvider {
    /// All known members
    members: Arc<RwLock<HashMap<String, Member>>>,
    /// The ID of the current node
    self_id: RwLock<Option<String>>,
    /// Channel for membership events
    event_tx: mpsc::Sender<MembershipEvent>,
    /// Receive side for tests
    event_rx: RwLock<Option<mpsc::Receiver<MembershipEvent>>>,
}

impl MemoryProvider {
    /// Create a new memory provider
    pub fn new() -> Self {
        let (tx, rx) = mpsc::channel(100);
        
        Self {
            members: Arc::new(RwLock::new(HashMap::new())),
            self_id: RwLock::new(None),
            event_tx: tx,
            event_rx: RwLock::new(Some(rx)),
        }
    }

    /// Start the heartbeat monitor
    pub fn start_heartbeat_monitor(
        self: Arc<Self>, 
        config: ClusterConfig
    ) -> tokio::task::JoinHandle<()> {
        let members = self.members.clone();
        let event_tx = self.event_tx.clone();
        
        tokio::spawn(async move {
            let mut interval = time::interval(Duration::from_secs(config.heartbeat_interval));
            
            loop {
                interval.tick().await;
                
                // Check member statuses
                let now = SystemTime::now();
                let mut to_mark_unreachable = Vec::new();
                let mut to_mark_down = Vec::new();
                
                {
                    // Read lock scope
                    let members_lock = members.read().unwrap();
                    
                    for (id, member) in members_lock.iter() {
                        if member.status == MemberStatus::Alive {
                            // Calculate time since last heartbeat
                            if let Ok(elapsed) = now.duration_since(member.last_heartbeat) {
                                if elapsed.as_secs() > config.unreachable_timeout {
                                    to_mark_unreachable.push(id.clone());
                                }
                            }
                        } else if member.status == MemberStatus::Unreachable {
                            if let Ok(elapsed) = now.duration_since(member.last_heartbeat) {
                                if elapsed.as_secs() > config.down_timeout {
                                    to_mark_down.push(id.clone());
                                }
                            }
                        }
                    }
                }
                
                // Update statuses with write lock
                {
                    let mut members_lock = members.write().unwrap();
                    
                    // Mark members as unreachable
                    for id in to_mark_unreachable {
                        if let Some(member) = members_lock.get_mut(&id) {
                            member.status = MemberStatus::Unreachable;
                            let event = MembershipEvent::MemberUnreachable(super::membership::MemberUnreachable {
                                member: member.clone(),
                            });
                            let _ = event_tx.try_send(event);
                            info!("Member {} marked as unreachable", id);
                        }
                    }
                    
                    // Mark members as down
                    for id in to_mark_down {
                        if let Some(member) = members_lock.get_mut(&id) {
                            member.status = MemberStatus::Down;
                            let event = MembershipEvent::MemberDown(super::membership::MemberDown {
                                member: member.clone(),
                            });
                            let _ = event_tx.try_send(event);
                            info!("Member {} marked as down", id);
                        }
                    }
                }
            }
        })
    }
}

#[async_trait]
impl ClusterProvider for MemoryProvider {
    async fn register(&self, config: &ClusterConfig) -> Result<(), ClusterError> {
        // Generate a unique ID for this node
        let id = format!("{}:{}:{}", config.name, config.host, config.port);
        
        // Create the member record with role and metadata
        let mut member = Member::new(id.clone(), config.host.clone(), config.port)
            .with_role(&config.name);
            
        // Add any custom metadata
        member.metadata.insert("version".to_string(), env!("CARGO_PKG_VERSION").to_string());
        member.metadata.insert("system_type".to_string(), "protoactor-rs".to_string());
        
        // Update status and timestamp
        member.status = MemberStatus::Alive;
        member.last_heartbeat = SystemTime::now();
        
        // Store in our members list
        {
            let mut members = self.members.write().unwrap();
            members.insert(id.clone(), member.clone());
            
            // Store self ID
            let mut self_id = self.self_id.write().unwrap();
            *self_id = Some(id.clone());
        }
        
        // Emit member joined event
        let event = MembershipEvent::MemberJoined(super::membership::MemberJoined {
            member: member.clone(),
        });
        
        let _ = self.event_tx.try_send(event);
        
        info!("Node registered with ID: {}", id);
        Ok(())
    }
    
    async fn deregister(&self) -> Result<(), ClusterError> {
        // Get our ID
        let id = {
            let self_id = self.self_id.read().unwrap();
            match &*self_id {
                Some(id) => id.clone(),
                None => return Err(ClusterError::ProviderError("Node not registered".to_string())),
            }
        };
        
        // Get our member record and update status
        let member = {
            let mut members = self.members.write().unwrap();
            
            match members.get_mut(&id) {
                Some(member) => {
                    member.status = MemberStatus::Left;
                    member.clone()
                },
                None => return Err(ClusterError::MemberNotFound(id)),
            }
        };
        
        // Emit member left event
        let event = MembershipEvent::MemberLeft(super::membership::MemberLeft {
            member,
        });
        
        let _ = self.event_tx.try_send(event);
        
        info!("Node deregistered: {}", id);
        Ok(())
    }
    
    async fn get_members(&self) -> Result<Vec<Member>, ClusterError> {
        let members = self.members.read().unwrap();
        let members = members.values().cloned().collect();
        Ok(members)
    }
    
    async fn monitor_members(&self) -> mpsc::Receiver<MembershipEvent> {
        let mut rx_opt = self.event_rx.write().unwrap();
        match rx_opt.take() {
            Some(rx) => rx,
            None => {
                // Create a new channel if the first one was already taken
                let (tx, rx) = mpsc::channel(100);
                self.event_tx.clone();
                rx
            }
        }
    }
    
    fn as_any(&self) -> &dyn Any {
        self
    }
} 